package com.learn.transform;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.concurrent.TimeUnit;

/**
 * @create: 2023-04-18 11:29
 * @author: Mr.Du
 * --------------
 * @notes:
 * 自定义一个dataSource
 * 两个流都消费该datasource，但是其中一个流对datasource进行额外处理
 * 然后将两个流connect
 **/
public class ConnectDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<Long> longDataStreamSource = env.addSource(new MyNoParallelSource());
        DataStreamSource<Long> longDataStreamSource2 = env.addSource(new MyNoParallelSource());

        SingleOutputStreamOperator<String> strDataStreamSource = longDataStreamSource2.map(new MapFunction<Long, String>() {
            @Override
            public String map(Long aLong) throws Exception {
                return "str_" + aLong;
            }
        });

        ConnectedStreams<Long, String> connectedStreams = longDataStreamSource.connect(strDataStreamSource);

        connectedStreams.map(new CoMapFunction<Long, String, Object>() {
            @Override
            public Object map1(Long l) throws Exception {
                return l;
            }

            @Override
            public Object map2(String s) throws Exception {
                return s;
            }
        }).print();

        env.execute("");
    }
}

class MyNoParallelSource implements SourceFunction<Long> {
    //定义一个变量，是否循环生成数据
    private boolean isRunning = true;
    private Long count = 0L;

    /**
     * 这是主要的方法，启动一个数据源
     * 实现数据的生成操作
     * @param ctx
     * @throws Exception
     */
    @Override
    public void run(SourceContext<Long> ctx) throws Exception {
        //不断生成订单数据
        while (isRunning){
            count+=1;
            //收集数据返回
            ctx.collect(count);

            //每隔一秒钟生成一条订单数据
            TimeUnit.SECONDS.sleep(1);
        }
    }

    /**
     * 取消数据的生成操作
     */
    @Override
    public void cancel() {
        isRunning = false;
    }
}
